package com.anmi.transfer.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import org.slf4j.MDC;
import org.springframework.util.Assert;

public abstract class AbstractCanalClient extends BaseCanalClient {

  public AbstractCanalClient() {}

  public AbstractCanalClient(String destination) {
    this(destination, null);
  }

  public AbstractCanalClient(String destination, CanalConnector connector) {
    this.destination = destination;
    this.connector = connector;
  }

  protected void start() {
    Assert.notNull(connector, "connector is null");
    thread = new Thread(this::process);
    thread.setUncaughtExceptionHandler(handler);
    running = true;
    thread.start();
  }

  protected void stop() {
    if (!running) {
      return;
    }
    running = false;
    if (thread != null) {
      try {
        thread.join();
      } catch (InterruptedException e) {
        // ignore
      }
    }
    MDC.remove("destination");
  }

  protected abstract void process();
}
